#include <UTILS/SBuffer.h>
#include <PECAgent/PEContext.h>
#include <DPS/PH/DpsFunctions.h>
#include <DPS/PH/DpsOPContext.h>
#include "UDOP_ReadPARDUMP.h"
#include <stdio.h>
#include "rngs.h"

SPC_NAMESPACE_USE;
UTILS_NAMESPACE_USE;
using namespace DPS;
using namespace std;
using namespace estd;

/*
 * Function used to initialize the operator
 */
void UDOP_ReadPARDUMP::initializeOperator() 
{
  // Create a user defined metric
  // opm.createMetric("intuples","Number of ingested tuples");
}

/*
 * Function used to finalize the operator
 */
void UDOP_ReadPARDUMP::finalizeOperator() 
{
}

/*
 * Function used to process command line arguments
 */
void UDOP_ReadPARDUMP::processCmdArgs(const string &args) 
{
  //SPCDBG(L_INFO, "Process command line arguments...", "dpsop");
  // ArgContainer arg(args); // include "ArgContainer.h"
  // arg.argc and arg.argv are now available for use
  // do other initialization here
}


/*
 * Process function for this source operator
 */
void UDOP_ReadPARDUMP::processInput0(const IPort0_t & tuple)
{
	int info = 0;
	int i = 1;
	int numpar = 0;
	int streamid = 0;
	char str1[50];
	char str2[100];
	char parfile[180];
	double smass = 0.0;
	float stlat = 0.0;
	float stlon = 0.0;
	float szlvl = 0.0;
	float ssigh = 0.0;
	float ssigw = 0.0;
	float ssigv = 0.0;
	int16_t spage = 0;
	int16_t shdwp = 0;
	int16_t sptyp = 0;
	int16_t spgrd = 0;
	int8_t sstid = 0;
	bool flag = true;
	const DpsOPContext &context = this->getContext();
	const std::string &args = context.getCmdArguments();
	//SPCDBG(L_INFO, "args is " << args, "dpsop");
	strcpy(str1, args.c_str());
	sscanf(str1,"%*s %*c%d%*c %*s %*c%d%*c",&streamid,&astep);
	sstid = streamid - 1;
	info = std::sprintf(parfile, "/home/shiming1/InfoSphereStreams/letkf-spade/hysplit/data/tes%d", sstid);
	OPort0_t otuple;

	FILE *fp;
	fp = fopen(parfile, "r");
	fgets(str2, 100, fp);
	fgets(str2, 100, fp);
	fgets(str2, 100, fp);
	fgets(str2, 100, fp);
	info = std::fscanf(fp,"%d", &numpar);
	fgets(str2, 100, fp);
	
	dlist<double> mass = DPS::makeDoubleList();
	dlist<float>  tlat = DPS::makeFloatList();
	dlist<float>  tlon = DPS::makeFloatList();
	dlist<float>  zlvl = DPS::makeFloatList();
	dlist<float>  sigh = DPS::makeFloatList();
	dlist<float>  sigw = DPS::makeFloatList();
	dlist<float>  sigv = DPS::makeFloatList();
	dlist<int16_t>  page = DPS::makeShortList();
	dlist<int16_t>  hdwp = DPS::makeShortList();
	dlist<int16_t>  ptyp = DPS::makeShortList();
	dlist<int16_t>  pgrd = DPS::makeShortList();
	//dlist<int8_t> stid = DPS::makeByteList();
 	tstep++;
//	double drand1 = 0.0; double drand2 = 0.0;

    if ((numpar > 0) && (tstep%astep==0) )
    {
	  for (i=0; i<numpar; i++)
	  {
		info = std::fscanf(fp,"%lE", &smass);
		info = std::fscanf(fp,"%f", &stlat);
		info = std::fscanf(fp,"%f", &stlon);
		info = std::fscanf(fp,"%f", &szlvl);
		info = std::fscanf(fp,"%f", &ssigh);
		info = std::fscanf(fp,"%f", &ssigw);
		info = std::fscanf(fp,"%f", &ssigv);
		info = std::fscanf(fp,"%hd", &spage);
		info = std::fscanf(fp,"%hd", &shdwp);
		info = std::fscanf(fp,"%hd", &sptyp);
		info = std::fscanf(fp,"%hd", &spgrd);
		info = std::fscanf(fp,"%d", &info);

		mass.push_back(smass);
		tlat.push_back(stlat);
		tlon.push_back(stlon);
		zlvl.push_back(szlvl);
		sigh.push_back(ssigh);
		sigw.push_back(ssigw);
		sigv.push_back(ssigv);
		page.push_back(spage);
		hdwp.push_back(shdwp);
		ptyp.push_back(sptyp);
		pgrd.push_back(spgrd);
		// Boolean flag is still true;
		flag = true;
	}
  }	
  else // if numpar == 0 or not analysis step
  {
		flag = false;
  }

		otuple.set_stid(sstid);
		otuple.set_mass(mass);
		otuple.set_tlat(tlat);
		otuple.set_tlon(tlon);
		otuple.set_zlvl(zlvl);
		otuple.set_sigh(sigh);
		otuple.set_sigw(sigw);
		otuple.set_sigv(sigv);
		otuple.set_page(page);
		otuple.set_hdwp(hdwp);
		otuple.set_ptyp(ptyp);
		otuple.set_pgrd(pgrd);
		otuple.set_flag(flag);
		
	fclose(fp);
	submit0(otuple);
}


/*
 * Function used to notify the operator about the readiness of all ports (this
 * is useful when submitting tuples from within a context other than the process
 * functions of the operator).
 */ 
/*
void UDOP_ReadPARDUMP::allPortsReady() 
{
} 
*/

/*
 * Terminate function, called asynchronously
 */
/*
void UDOP_ReadPARDUMP::prepareToTerminate() 
{
}
*/

/*
 * Function used to update performance counters
 */
/*
static void updatePerformanceCounters(DpsOPMetrics & opm) 
{ 
  static long long ntuples = 0;
  // This is not the ideal way of using performance counters
  // Ideally one should store the result from getMetricByName, 
  // in a class member variable. For instance:
  //   DpsMetric & m = opm.getMetricByName("intuples");
  // m could be made a member variable of UDOP_ReadPARDUMP
  // Then, the following is a more efficient way to use the counter:
  //   m.setValue(ntuples++);
  opm.getMetricByName("intuples").setValue(ntuples++);
}
*/

/*
 * Checkpointing function for saving state, called asynchronously 
 */
/*
void UDOP_ReadPARDUMP::getCheckpoint(SBuffer & checkpoint) 
{
  SPCDBG(L_DEBUG, "Getting operator state...", "dpsop");
}
*/

/*
 * Checkpointing function for restoring state, called asynchronously
 */
/*
void UDOP_ReadPARDUMP::restoreCheckpoint(SBuffer & checkpoint) 
{
  SPCDBG(L_DEBUG, "Restoring operator state...", "dpsop");
}
*/


